package com.atguigu.app.dwd.db;

import com.atguigu.common.Constant;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

//数据流：web/app -> Nginx -> 业务服务器(Mysql) -> Maxwell -> Kafka(ODS) -> FlinkApp -> Kafka(DWD)
//程  序：Mock -> maxwell.sh -> Kafka(ZK) -> Dwd03_TradeCartAdd -> Kafka(ZK)
public class Dwd03_TradeCartAdd {

    public static void main(String[] args) {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  //生产环境中,主题并行度与Kafka主题的分区数保持一致
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //Logger logger = LoggerFactory.getLogger(Dwd01_TrafficBaseLogSplit.class);
        //logger.info("aaa");

        //1.1 开启CheckPoint
        //env.enableCheckpointing(60000 * 5);
        //env.setStateBackend(new HashMapStateBackend());

        //1.2 CheckPoint相关设置
        //CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        //checkpointConfig.setCheckpointTimeout(10000L);
        //checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/flink-ck");
        //Cancel任务时保存最后一次CheckPoint结果
        //checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //checkpointConfig.setMinPauseBetweenCheckpoints(5000L);
        //设置重启策略
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 1000L));

        //TODO 2.读取Kafka topic_db 主题数据创建动态表
        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd03_cart_info_230524"));

        //TODO 3.过滤出加购表数据
        Table cartInfoTable = tableEnv.sqlQuery("" +
                "select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['sku_id'] sku_id,\n" +
                "    `data`['cart_price'] cart_price,\n" +
                "    if(`type` = 'insert',`data`['sku_num'],cast(cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int) as string)) sku_num,\n" +
                "    `data`['sku_name'] sku_name,\n" +
                "    `data`['is_checked'] is_checked,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    `data`['operate_time'] operate_time,\n" +
                "    `data`['is_ordered'] is_ordered,\n" +
                "    `data`['order_time'] order_time,\n" +
                "    `data`['source_type'] source_type,\n" +
                "    `data`['source_id'] source_id\n" +
                "from ods_topic_db\n" +
                "where `database` = 'gmall-230524-flink'\n" +
                "and `table` = 'cart_info'\n" +
                "and (`type` = 'insert' or (`type` = 'update' and `old`['sku_num'] is not null and cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int)))");
        tableEnv.createTemporaryView("cart_info", cartInfoTable);

        //TODO 4.构建Kafka Sink Table
        tableEnv.executeSql("" +
                "create table dwd_cart_info(\n" +
                "    `id` string,\n" +
                "    `user_id` string,\n" +
                "    `sku_id` string,\n" +
                "    `cart_price` string,\n" +
                "    `sku_num` string,\n" +
                "    `sku_name` string,\n" +
                "    `is_checked` string,\n" +
                "    `create_time` string,\n" +
                "    `operate_time` string,\n" +
                "    `is_ordered` string,\n" +
                "    `order_time` string,\n" +
                "    `source_type` string,\n" +
                "    `source_id` string\n" +
                ")" + KafkaUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));

        //TODO 5.将数据写出到Kafka
        tableEnv.executeSql("insert into dwd_cart_info select * from cart_info");

    }
}
